#include "../include/Debug.h"
#include "../include/AsynchIO.h"
#include "../include/InetAddr.h"
#include "../include/TaskWithMQ.h"
#include "../include/EpollProactor.h"
#include "../include/ThreadManager.h"
#include "../include/AsynchAcceptor.h"

#include "../logger/Logger.h"

#include <assert.h>
#include <signal.h>

// log server
THREAD_FUNC_RETURN_T event_loop (void *arg)
{
    Proactor *p = reinterpret_cast<Proactor *> (arg);
    p->run_proactor_event_loop ();
    return 0;
}
// --------------------------------------------------------
enum
{
    MSG_LOGIN           = 200,
    MSG_LOGIN_RESP      = 201,
    MSG_HEART_BEAT      = 202
};
struct MsgHeader
{
    int msgid;
    int msglen;  // not include header
    MsgHeader ()
    : msgid (-1)
    , msglen (0)
    {}
};
struct LoginInfo
{
    MsgHeader header;
    //
    int  area_id;
    int  service_id;
    char username[16];
    char userpasswd[16];
    char reserve[8];

    LoginInfo ()
    : area_id (0)
    , service_id (0)
    {
        memset (username, 0, sizeof (username));
        memset (userpasswd, 0, sizeof (userpasswd));
        memset (reserve, 0, sizeof (reserve));
    }
};
struct LoginReturnInfo
{
    MsgHeader header;
    //
    int  result;
    int  userid;
    char reserve[8];
};
struct HeartBeatInfo
{
    MsgHeader header;
    int userid;
};
class RecvMessageHandler;
class EventItem
{
public:
    EventItem ()
    : sock_handler (0)
    , message_block (0)
    {
    }
    ~EventItem ()
    {
        if (this->message_block)
        {
            this->message_block->release ();
        }
    }
    RecvMessageHandler *sock_handler;
    MessageBlock *message_block;
};
#define INIT_BUFF_LEN          4096 
static Logger *netrecv_log = LogManager::instance ()->get_logger ("server.netrecv");
class RecvMessageHandler : public AsynchHandler
{
public:
    RecvMessageHandler () 
    : read_stream_buff_ (0)
    , write_stream_buff_ (0)
    , bytes_read_ (0)
    , bytes_write_ (0)
    , write_count_ (0)
    , closed_ (0)
    , stop_read_ (0)
    , stop_write_ (0)
    {
    }
    ~RecvMessageHandler ()
    { 
    }
    virtual int open (NDK_HANDLE handle)
    {
        char buff[32] = {0};
        this->remote_addr_.addr_to_string (buff, 32);
        netrecv_log->rinfo ("accept a new connection [%s]", buff);
        this->read_stream_buff_ = new MessageBlock (INIT_BUFF_LEN);
        if (this->read_stream_.open (this, handle) == 0)
        {
            if (this->start_read (0) != 0)
            {
                netrecv_log->error ("start read faild");
                return -1;
            }
        }else
            netrecv_log->error ("read stream open faild");
        if (this->write_stream_.open (this, handle) != 0)
        {
            netrecv_log->error ("write stream open faild");
            return -1;
        }
        return 0;
    }
    virtual void remote_addr (const InetAddr &remote_addr)
    {
        this->remote_addr_ = remote_addr;
    }
    int start_read (int reset)
    {
        if (reset)
            this->read_stream_buff_->reset ();
        return this->read_stream_.read (this->read_stream_buff_, 
                this->read_stream_buff_->size () - 
                this->read_stream_buff_->length ());
    }
    int start_write (MessageBlock *mb)
    {
        Guard_T<ThreadMutex> g (this->write_mutex_);
        int result = this->write_stream_.write (mb, mb->size ());
        if (result == 0) ++this->write_count_;
        return result;
    }
    virtual int handle_close (const int mask)
    {
        if (mask == AsynchHandler::READ_MASK)
        {
            NDK_OS::closesocket (this->handle ());
            netrecv_log->rinfo ("recv data length = %d", this->bytes_read_);
            if (this->read_stream_buff_ == 0)
                delete this->read_stream_buff_;
            this->read_stream_buff_ = 0;
            this->stop_read_ = 1;
        }else if (mask == AsynchHandler::WRITE_MASK)
        {
            netrecv_log->rinfo ("send data length = %d", this->bytes_write_);
            if (this->write_stream_buff_ == 0)
                delete this->write_stream_buff_;
            this->write_stream_buff_ = 0;
            this->stop_write_ = 1;
        }
        if (this->stop_read_ && this->stop_write_)
        {
            char buff[32] = {0};
            this->remote_addr_.addr_to_string (buff, sizeof (buff));
            netrecv_log->debug ("release handler:%s", buff);
            delete this;
        }
        return 0;
    }
    virtual int handle_read_stream (const AsynchReadStream::Result &result);
    virtual int handle_write_stream (const AsynchWriteStream::Result &result);
protected:
    AsynchReadStream read_stream_;

    AsynchWriteStream write_stream_;

    ThreadMutex   write_mutex_;

    MessageBlock *read_stream_buff_;
    MessageBlock *read_stream_tail_;

    MessageBlock *write_stream_buff_;

    InetAddr remote_addr_;

    int bytes_read_;

    int bytes_write_;

    int write_count_;

    int closed_;

    int stop_read_;

    int stop_write_;
};
static int g_userid = 0;
static Logger *msghandle_log = LogManager::instance ()->get_logger ("server.msghandle_task");
class MsgHandleTask : public TaskWithMQ
{
public:
    MsgHandleTask ()
    : TaskWithMQ (ThreadManager::instance ())
    {
    }
    
    ~MsgHandleTask (){}
    virtual int open (void * = 0)
    {
        return this->activate (Thread::THR_JOIN, 2);
    }
    virtual int svc ()
    {
        MessageBlock *mb = 0;
        TimeValue tv (3, 0);
        int count = 0;
        while (1)
        {
            count = this->getq_n (mb, 128, &tv);
            if (mb == 0 || count == 0) continue;
            msghandle_log->trace ("task get %d message block!", count);

            MessageBlock *tmb = mb;
            for (; mb != 0; mb = mb->next ())
            {
                EventItem *pitem = reinterpret_cast<EventItem *>(mb->base ());
                if (pitem)
                {
                    MsgHeader *header = 
                        reinterpret_cast<MsgHeader *>(pitem->message_block->base ());
                    if (header->msgid == MSG_LOGIN)
                    {
                        LoginInfo *loginfo = 
                            reinterpret_cast<LoginInfo*>(pitem->message_block->base ());
                        MessageBlock *msg = new MessageBlock (
                                sizeof (LoginReturnInfo));
                        msg->wr_ptr (sizeof (LoginReturnInfo));
                        LoginReturnInfo *return_info 
                            = reinterpret_cast<LoginReturnInfo *>(msg->base ());
                        return_info->header.msgid  = MSG_LOGIN_RESP;
                        return_info->header.msglen = sizeof (LoginReturnInfo) - 
                            sizeof (MsgHeader);
                        if (strcmp (loginfo->userpasswd, "passwd") == 0
                                && strcmp (loginfo->reserve, "911") == 0)
                        {
                            return_info->result = 0;
                            return_info->userid = ++g_userid;
                            msghandle_log->rinfo ("user:%s login successfully!",
                                    loginfo->username);
                        }
                        else
                        {
                            return_info->result = 1001;
                            msghandle_log->error ("user:%s login failed!",
                                    loginfo->username);
                        }
                        pitem->sock_handler->start_write (msg);
                    }else if (header->msgid == MSG_HEART_BEAT)
                    {
                        HeartBeatInfo  *hbinfo = 
                            reinterpret_cast<HeartBeatInfo*>(pitem->message_block->base ());
                        msghandle_log->rinfo ("user:%d's heart beat!",
                                hbinfo->userid);
                    }
                    delete pitem;
                }
            }
            tmb->release ();
        }
        return 0;
    }
protected:
};
typedef Singleton_T<MsgHandleTask> msg_handle_task;
// +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
int RecvMessageHandler::handle_read_stream (const AsynchReadStream::Result &result)
{
    if (!result.success ())
    {
        netrecv_log->error ("read stream not success, errno = %d",
                result.error ());
        if (result.error () != EAGAIN)
        {
            return -1;
        }
    }else if (result.bytes_transferred () == 0)
    {
        char buff[32] = {0};
        this->remote_addr_.addr_to_string (buff, sizeof (buff));
        netrecv_log->error ("client:%s closed", buff);
        return -1;
    }
    // read header.
    MessageBlock *message_block = result.message_block ();

    int next_package = 0;
    while (1)
    {
        if (message_block->length () > sizeof (MsgHeader))
        {
            MsgHeader *mh = reinterpret_cast<MsgHeader*>(message_block->base ());
            int package_len = mh->msglen + sizeof (MsgHeader);
            if (message_block->length () >= package_len)
                // recv a complete package
            {
                MessageBlock *mb = new MessageBlock(package_len);
                memcpy (mb->wr_ptr (), message_block->rd_ptr (), package_len);
                message_block->rd_ptr (package_len);
                mb->wr_ptr (package_len);
                EventItem *item = new EventItem;
                item->sock_handler = this;
                item->message_block = mb;
                msg_handle_task::instance ()->putq (new MessageBlock ((char*)item));
                netrecv_log->debug ("recv a complete package");
                next_package = 1;
            }else
            {
                if (package_len > message_block->size ())
                {
                    this->read_stream_buff_ = new MessageBlock (package_len);
                    memcpy (this->read_stream_buff_->wr_ptr (),
                            message_block->rd_ptr (),
                            message_block->length ());
                    this->read_stream_buff_->wr_ptr (message_block->length ());
                    delete message_block;
                    netrecv_log->debug ("recv a bigger package");
                    return this->start_read (0);
                }else
                {
                    int l = message_block->length ();
                    memmove (this->read_stream_buff_->base (), 
                            message_block->rd_ptr (), 
                            message_block->length ());
                    message_block->reset ();
                    message_block->wr_ptr (l);
                    netrecv_log->debug ("split an uncomplete package, body is not complete");
                    return this->start_read (0);
                }
            }
        }else
        {
            if (!next_package)
            {
                netrecv_log->debug ("the first package is uncompleted, header is not complete");
                return this->start_read (0);
            }else
            {
                if (message_block->length () == 0)
                {
                    netrecv_log->debug ("split some completed package");
                    return this->start_read (1);
                }
                else  // > 0
                    // @point 1: 这说明是收到了多个包。
                {
                    int l = message_block->length ();
                    memmove (this->read_stream_buff_->base (), 
                            message_block->rd_ptr (), 
                            message_block->length ());
                    message_block->reset ();
                    message_block->wr_ptr (l);
                    netrecv_log->debug ("split an uncomplete package, header is not complete");
                    return this->start_read (0);
                }
            }
        }
    }
    return 0;
}
int RecvMessageHandler::handle_write_stream (const AsynchWriteStream::Result &result)
{
    --this->write_count_;
    if (!result.success ())
    {
        netrecv_log->error ("write stream not success, errno = %d",
                result.error ());
        if (result.error () != EAGAIN)
        {
            return -1;
        }
    }
    this->bytes_write_ += result.message_block ()->length ();
    result.message_block ()->release ();
    if (this->write_count_ == 0)
        this->stop_write_ = 1;
    else
        this->stop_write_ = 0;
    return 0;
}
int main (int argc, char *argv[])
{
    int port = 8881;
    EpollProactor *ep = new EpollProactor ();
    if (ep->open () != 0)
    {
        fprintf (stderr, "epoll proactor open failed\n");
        return -1;
    }
    Proactor *p = new Proactor (ep, 1);
    Proactor::instance (p);
    ThreadManager::instance ()->spawn_n (1, event_loop, Proactor::instance ());

    LogManager::instance ()->init ("slog.ini");
    msg_handle_task::instance ()->open ();
    //msg_handle_task::instance ()->open ();
    AsynchAcceptor<RecvMessageHandler> acceptor;
    InetAddr local_addr (port);
    assert (Proactor::instance () != 0);
    if (acceptor.open (local_addr,
                Proactor::instance (),
                1,
                1024 * 1024 * 2) != 0)
    {
        fprintf (stderr, "accept open failed\n");
        return 0;
    }
    ThreadManager::instance ()->wait ();
    return 0;
}

